package com.jd.ssdb4j.stream;

import com.jd.ssdb4j.base.Cmd;
import com.jd.ssdb4j.base.Response;
import com.jd.ssdb4j.exception.SSDBException;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public abstract class AbstractSSDBStream implements SSDBStream {

    protected Socket socket;
    protected MemoryStream input;

	public Response req(Cmd cmd, byte[]... vals) {

		beforeExec();

        try {

            ArrayList<byte[]> list = new ArrayList<byte[]>();

            for (byte[] s : vals) {

                list.add(s);
            }

            return this.request(cmd.name(), list);
		} catch (Throwable e) {

			return whenError(e);
		}
	}

    public Response request(String cmd, List<byte[]> params) throws Exception{

        MemoryStream buf = new MemoryStream(4096);
        Integer len = cmd.length();

        buf.write(len.toString());
        buf.write('\n');
        buf.write(cmd);
        buf.write('\n');

        for(byte[] bs : params){

            len = bs.length;
            buf.write(len.toString());
            buf.write('\n');
            buf.write(bs);
            buf.write('\n');
        }
        buf.write('\n');

        send(buf);
        List<byte[]> list = recv();

        return new Response(list);
    }

    private void send(MemoryStream buf) throws Exception{
        OutputStream os = socket.getOutputStream();
        os.write(buf.buf, buf.data, buf.size);
        os.flush();
    }

    private List<byte[]> recv() throws Exception{
        input.nice();
        InputStream is = socket.getInputStream();
        while(true){
            List<byte[]> ret = parse();
            if(ret != null){
                return ret;
            }
            byte[] bs = new byte[8192];
            int len = is.read(bs);
            input.write(bs, 0, len);
        }
    }

    private List<byte[]> parse(){
        ArrayList<byte[]> list = new ArrayList<byte[]>();
        byte[] buf = input.buf;

        int idx = 0;
        while(true){
            int pos = input.memchr('\n', idx);
            if(pos == -1){
                break;
            }
            if(pos == idx || (pos == idx + 1 && buf[idx] == '\r')){
                if(list.isEmpty()){
                    idx += 1;
                    continue;
                }else{
                    input.decr(idx + 1);
                    return list;
                }
            }
            String str = new String(buf, input.data + idx, pos - idx);
            int len = Integer.parseInt(str);
            idx = pos + 1;
            if(idx + len >= input.size){
                break;
            }
            byte[] data = Arrays.copyOfRange(buf, input.data + idx, input.data + idx + len);
            idx += len + 1;
            list.add(data);
        }
        return null;
    }

	protected void beforeExec() {
	}

	protected void beforeReturn(Response resp) {
	}

	protected Response whenError(Throwable e) {
		throw new SSDBException(e);
	}
}
